package cn.chenchendemospringboot.order_demo.service.impl;

import cn.chenchendemospringboot.order_demo.entity.KafkaEventPublish;
import cn.chenchendemospringboot.order_demo.mapper.KafkaEventPublishMapper;
import cn.chenchendemospringboot.order_demo.service.KafkaEventPublishService;
import cn.chenchendemospringboot.order_demo.utils.KafkaUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;

@Service
@Slf4j
public class KafkaEventPublishServiceImpl implements KafkaEventPublishService {

    @Autowired
    KafkaEventPublishMapper kafkaEventPublishMapper;



    @Override
    public boolean save(KafkaEventPublish kafkaEventPublish) {
        return kafkaEventPublishMapper.insert(kafkaEventPublish) > 0;
    }

    @Override
    public List<KafkaEventPublish> listByCondition(String status) {
        QueryWrapper<KafkaEventPublish> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", status);
        return kafkaEventPublishMapper.selectList(queryWrapper);
    }

    @Override
    public boolean update(KafkaEventPublish kafkaEventPublish) {
        return kafkaEventPublishMapper.updateById(kafkaEventPublish) > 0;
    }

    @Override
    public boolean updateByIds(List<Long> ids, String status) {
        UpdateWrapper<KafkaEventPublish> updateWrapper = new UpdateWrapper<>();
        updateWrapper.in("id", ids).set("status", status);
        return kafkaEventPublishMapper.update(null,updateWrapper)>0;
    }

    /**
     * 定时任务轮询发布
     * 查询所有状态为NEW的时间
     * 发送到消息队列   发送的是订单的json串  即 payload字段
     * 更改消息状态
     */
    @Transactional(rollbackFor = Exception.class)
    public void publish() {
        List<KafkaEventPublish> publisList = listByCondition("NEW");
        if(!CollectionUtils.isEmpty(publisList)){
            List<Long> longs = sendEventPublish(publisList);
            if(!CollectionUtils.isEmpty(longs)){
                boolean news = updateByIds(longs, "Publish");
            }
        }

    }

    private static List<Long> sendEventPublish(List<KafkaEventPublish> eventPublishes) {
        if (CollectionUtils.isEmpty(eventPublishes)) {
            return Collections.emptyList();
        }
        List<Long> ids = Lists.newArrayList();
        for (KafkaEventPublish eventPublish : eventPublishes) {
            try {
                KafkaUtils.sendSync(eventPublish.getEventType(), eventPublish.getPayload());
                ids.add(eventPublish.getId());
            } catch (Exception e) {
                log.error("发送Kafka消息失败，eventPublish={}", eventPublish, e);
            }
        }
        log.debug("发送Kafka消息成功，ids={}", ids);
        return ids;
    }
}
